package com.dxy.client.client;

import com.alibaba.fastjson.JSON;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @Description socket客户端服务
 * @Author dangxianyue
 * @Date 2023/4/1 14:53
 */
@Slf4j
@Data
public class SocketClient {

    private static final String DELIMITER = "@@";

    private String serverIp;
    private int serverPort;
    private String name = "xxx";
    // 读取超时时间，单位是秒
    private long timeout = 10;
    private TimeUnit timeoutUnit = TimeUnit.SECONDS;
    private Socket socket;
    private BufferedReader br;
    private PrintWriter pw;

    /**
     * 把异步转化成同步，实现“请求-响应”同步通信模型的核心类
     * <p>
     * “等待-通知”模型常见的有三种方式实现：
     * 一：synchronized + wait / notify / notifyAll
     * 很难控制通知到具体的线程
     * 用notify只能随机唤醒一个在等待队列上的线程，有信号丢号的风险
     * 用notifyAll可以唤醒所有在等待队列上的线程，效率较低，此方案慎用
     * 二、CountDownLatch
     * 每个请求都用一个cdl去阻塞，当异步响应到来时，可以根据requestId找到对应cdl进行唤醒
     * 这样就可以实现通知到具体的线程
     * RocketMQ底层就是使用这个机制
     * 三、ReentrantLock + Condition
     * 条件锁机制，也可以通知到具体线程
     * Dubbo底层把Netty异步转化为同步就是用了此种机制，具体见 DefaultFuture 类
     * 参考：详解Java中异步转同步的六种方法：
     * https://www.zhangshengrong.com/p/AvN6Yoeeam/
     * <p>
     * 此处使用第二种方案：CountDownLatch + Future
     * 每次请求都对应着一个SyncFuture，因此请求和响应的数据包中都应该包含一个唯一对应的requestId
     * 在发送请求时，把 Map<requestId, SyncFuture> 缓存起来，然后发送消息，并cdl.await等待
     * 当异步响应到来时，根据响应报文中的requestId，从缓存找出对应的SyncFuture，进行cdl.countDown，
     * 这样通知到指定的线程去获取数据
     *
     * @param <T>
     */
    private static class SyncFuture<T> implements Future<T> {

        // 请求和响应是一一对应的，设置cdl初始化值为1
        private CountDownLatch cdl = new CountDownLatch(1);

        // 响应信息
        private T response;

        // 取消任务，如果取消任务成功则返回true，如果取消任务失败则返回false。
        // 参数mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务
        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            return false;
        }

        // 表示任务是否被取消成功，如果在任务正常完成前被取消成功，则返回 true
        @Override
        public boolean isCancelled() {
            return false;
        }

        // 表示任务是否已经完成，若任务完成，则返回true
        // 此处表示的是异步响应是否到来
        @Override
        public boolean isDone() {
            return this.response != null;
        }

        // 获取执行结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回
        // 此处表示的是，调用方在发送请求后，调用此方法获取响应结果进行返回
        @Override
        public T get() throws InterruptedException, ExecutionException {
            cdl.await();// 关键代码，阻塞等待
            return this.response;
        }

        // 获取执行结果，如果在指定时间内，还没获取到结果，就直接返回null
        @Override
        public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
            if (cdl.await(timeout, unit)) {
                return this.response;
            }
            return null;
        }

        // 用于设置响应结果，并且做countDown操作，通知请求线程
        // 当异步响应到来时，根据requestId获取到对应的SyncFuture，调用此方法
        public void setResponse(T response) {
            this.response = response;
            cdl.countDown();// 关键代码，countDown到0，上面await阻塞住的动作就会放行
        }
    }

    /**
     * Future的缓存器，把requestId、SyncFuture对应缓存起来，为了在异步响应到来时找到对应的cdl进行唤醒
     * 也可以使用ConcurrentHashMap实现，但由于socket异常会导致响应为null的情况，会让map中占有越来越多，因此需要实现过期淘汰的功能
     * guava自动帮我们实现了基于容量的清理、基于时间的清理等，使用方便。
     */
    private static final LoadingCache<String, SyncFuture<String>> futureCache = CacheBuilder.newBuilder()
            //设置缓存容器的初始容量为10
            .initialCapacity(100)
            // maximumSize 设置缓存大小
            .maximumSize(10000)
            //设置并发级别为20，并发级别是指可以同时写缓存的线程数
            .concurrencyLevel(20)
            // expireAfterWrite设置写缓存后8秒钟过期
            .expireAfterWrite(10, TimeUnit.SECONDS)
            //设置缓存的移除通知
            .removalListener(new RemovalListener<Object, Object>() {
                @Override
                public void onRemoval(RemovalNotification<Object, Object> notification) {
                    log.debug("LoadingCache: {} was removed, cause is {}", notification.getKey(), notification.getCause());
                }
            })
            //build方法中可以指定CacheLoader，在缓存不存在时通过CacheLoader的实现自动加载缓存
            .build(new CacheLoader<String, SyncFuture<String>>() {
                @Override
                public SyncFuture<String> load(String key) throws Exception {
                    // 当获取key的缓存不存在时，不需要自动添加
                    return null;
                }
            });

    /**
     * 接受消息线程池
     */
    private ExecutorService receiveDataThreadPool = Executors.newFixedThreadPool(1);

    /**
     * 连接服务器
     *
     * @return
     */
    public boolean connect() {
        try {
            socket = new Socket(serverIp, serverPort);
            br = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8"));
            pw = new PrintWriter(new OutputStreamWriter(socket.getOutputStream(), "UTF-8"));
            receiveDataThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    receive();
                }
            });
            log.info("[{}]socket连接成功", name);
            return true;
        } catch (IOException e) {
            log.error("[{}]socket连接异常", name, e);
            return false;
        }
    }

    /**
     * 接受消息
     */
    private void receive() {
        try {
            String line;
            // readLine会阻塞，直到遇到\r、\n、\r\n才返回
            // 当socket断开时，返回null，循环结束
            while ((line = br.readLine()) != null) {
                log.info("[{}]socket收到响应：{}", name, line);

                // 根据响应信息中的requestId，从缓存中获取对应的Future，唤醒线程
                String requestId = JSON.parseObject(line).getString("dataId");
                SyncFuture<String> syncFuture = futureCache.getIfPresent(requestId);

                // 如果不为null, 则通知返回
                if(syncFuture != null) {
                    syncFuture.setResponse(line);
                    //主动释放
                    futureCache.invalidate(requestId);
                }
            }
        } catch (Exception ignored) {
        }
        log.info("[{}]socket连接断开", name);
        close();
        connect();
    }

    /**
     * 关闭socket
     */
    private void close() {
        if (socket != null) {
            try {
                socket.shutdownInput();
                socket.shutdownOutput();
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 发送消息（实现“请求-响应”同步通信机制）
     * 消息格式为json，且每个数据包都需包含唯一的dataId，响应报文中也需携带此id
     * 如：{"dataId":"123456"}
     *
     * @param message
     */
    public String send(String message) {
        if (isServerEnable()) {
            log.info("[{}]socket发送消息：{}", name, message);

            // 获取唯一的requestId
            String requestId = JSON.parseObject(message).getString("dataId");

            // 缓存Futrue
            SyncFuture<String> future = new SyncFuture<>();
            futureCache.put(requestId, future);

            // 发送消息
            pw.print(message + DELIMITER);
            pw.flush();

            // 获取响应结果返回，此处会阻塞
            try {
                return future.get(timeout, timeoutUnit);
            } catch (InterruptedException | TimeoutException | ExecutionException e) {
                throw new RuntimeException(String.format("[%s]socket发送异常", name), e);
            }
        } else {
            throw new RuntimeException(String.format("[%s]socket服务不可用", name));
        }
    }

    /**
     * 判断socket是否可用
     *
     * @return
     */
    public boolean isSocketEnable() {
        if (
                socket != null
                        && socket.isBound()
                        && !socket.isClosed()
                        && socket.isConnected()
                        && !socket.isInputShutdown()
                        && !socket.isOutputShutdown()
        ) {
            try {
                //发送1个字节的紧急数据，默认情况下，服务器端没有开启紧急数据处理，不影响正常通信
                socket.sendUrgentData(0xFF);
                return true;
            } catch (Exception ignored) {
            }
        }

        return false;
    }

    /**
     * 判断服务器是否可用
     *
     * @return
     */
    public boolean isServerEnable() {
        return isSocketEnable() || connect();
    }
}
